/**
 * 
 */
package dm.ass2.user;

import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import dm.ass1.io.ProfileWritable;
import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.user.io.CBElement;
import dm.ass2.user.io.UserMapOutputValueWritable;

/**
 * @author yaron
 * 
 */
public class UserMap extends Mapper<LongWritable, Text, IntWritable, UserMapOutputValueWritable> {
	private static final Log LOG = LogFactory.getLog(UserMap.class);
	
	List<CodebookItemsWritable> cbItems;
	int numOfUserClusters;
	int numOfItemClusters;

	@Override
	protected void map(LongWritable key, Text userProfileString, Context context) throws IOException, InterruptedException {
		ProfileWritable userProfile = ProfileWritable.fromString(userProfileString.toString());
		int userId = userProfile.getRater();
		int[] items = userProfile.getRated();
		int[] ratings = userProfile.getRating();
		CBElement[] cbElements = new CBElement[numOfItemClusters];
		
		int bestClusterId  = -1; // initializing the user cluster.
		double bestClusterErr = Double.MAX_VALUE;  // initializing the minimum error.
		for (int j = 0; j < numOfUserClusters; j++) {
			double err = 0;

			for (int itemIndex = 0; itemIndex < items.length; itemIndex++) {
				CodebookItemsWritable cbItem = findRating(items[itemIndex]);
				if (cbItem == null) {
					continue;
				}
				
				double st = ratings[itemIndex];
				double b = cbItem.getUserClusterRating()[j];
				err += Math.pow((st-b), 2.0);
				
				CBElement cbElement = cbElements[cbItem.getItemClusterId()];
				if (cbElement == null) {
					cbElement = new CBElement(0, 0);
					cbElements[cbItem.getItemClusterId()] = cbElement;
				}
				cbElement.increment();
				cbElement.addToSum(st);
			}
			
			if (err < bestClusterErr) {
				bestClusterErr = err;
				bestClusterId = j;
			}
		}
		
		IntWritable keyOut = new IntWritable(bestClusterId);
		UserMapOutputValueWritable valueOut = new UserMapOutputValueWritable(new int[] {userId},cbElements);
		context.write(keyOut, valueOut);
	}

	private CodebookItemsWritable findRating(int item) {
		for (CodebookItemsWritable cbItem : cbItems) {
			if (ArrayUtils.contains(cbItem.getItemIds(), item)) {
				return cbItem;
			}
		}
		return null;
	}
	
	

	/**
	 * Initialization: loads the CodeBookItem to the memory.
	 */
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		
		this.numOfItemClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_ITEM_CLUSTERS));
		this.numOfUserClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_USER_CLUSTERS));

		Path cbItemsPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_ITEMS_FILENAME);
		FileSystem fs = FileSystem.get(context.getConfiguration());
		cbItems = new ArrayList<CodebookItemsWritable>();

		FSDataInputStream in = fs.open(cbItemsPath);
		try {
			CodebookItemsWritable cbItem = new CodebookItemsWritable();
			cbItem.readFields(in);
			while (cbItem != null) {
				cbItems.add(cbItem);
				cbItem = new CodebookItemsWritable();
				cbItem.readFields(in);				
			}
		} catch (EOFException e) {
			LOG.info(String.format("Setup: read %d records to cb item.", cbItems.size()));
		} finally {
			in.close();
		}
	}

}
